package com.fxd.kafka.controller;

import com.fxd.kafka.enums.KafkaTopicEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author fuxudong
 * @date 2022/08/02
 */
@Slf4j
@RestController
@RequestMapping("/easy")
public class EasyController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    // 发送消息
    @GetMapping("/kafka/{message}")
    public void sendMessage(@PathVariable("message") String message) {
        kafkaTemplate.send(KafkaTopicEnum.DEMO_TOPIC.getTopic(), message);
    }

    @GetMapping("/kafka/callback/{message}")
    public void sendMessage2(@PathVariable("message") String message) {
        kafkaTemplate.send(KafkaTopicEnum.DEMO_TOPIC.getTopic(), message).addCallback(success -> {
            // 消息发送到的topic
            String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            log.info("主题: {}, 分区: {}, offset: {} 发送消息成功", topic, partition, offset);
        }, fail -> {
            log.info("主题: {} 发送消息发生异常: {}", KafkaTopicEnum.DEMO_TOPIC.getTopic(), fail.getCause());
        });
    }

    @Transactional
    @GetMapping("/kafka/transaction/{message}")
    public void sendMessage3(@PathVariable("message") String message) {
        kafkaTemplate.send(KafkaTopicEnum.DEMO_TOPIC.getTopic(), message);
        System.out.println(1 / 0);
    }

}
